package com.hliushi.datasource.crawler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.hliushi.datasource.bean.CovidBean;
import com.hliushi.datasource.bean.DangerAreasBean;
import com.hliushi.datasource.utils.HttpUtils;
import com.hliushi.datasource.utils.TimeUtils;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @ClassName Covid19DataCrawler
 * @Description 实现疫情数据爬取
 * @Author hliushi
 * @Date 2021/7/1 8:16
 * @Version V1.0
 **/
// @RunWith(SpringJUnit4ClassRunner.class)
// @SpringBootTest(classes = DatasourceApplication.class)
@Component
public class Covid19DataCrawler {

    @Autowired
    private KafkaTemplate kafkaTemplate;


    // @Test
    // public void testKafkaTemplate() throws InterruptedException {
    //     kafkaTemplate.send("test", 1, "abc");
    //     Thread.sleep(1000000000);
    // }


    // 后续需要将该方法改为定时任务, 如每天8点定时爬取疫情数据
    // @Scheduled(cron = "0 0 8 * * ?")
    @Scheduled(initialDelay = 1000, fixedDelay = 1000 * 60 * 60 * 24) // 1秒之后执行, 然后每隔24小时在执行一次
    // @Scheduled(initialDelay = 1000, fixedDelay = 1000 * 60) // 1秒之后执行, 然后每隔一分钟执行一次
    public void testCrawling() throws Exception {
        String datetime = TimeUtils.format(System.currentTimeMillis(), "yyyy-MM-dd");

        // 1.爬取指定页面
        String html = HttpUtils.getHtml("https://ncov.dxy.cn/ncovh5/view/pneumonia");
        // System.out.println("html = " + html);

        // 2.解析页面你的指定内容 - 即id为getAreaStat标签中的全国疫情数据
        Document doc = Jsoup.parse(html);

        String text = doc.select("script[id=getAreaStat]").toString();
        System.out.println("text = " + text);

        // 3.使用正则表达式获取json格式的疫情数据
        String patten = "\\[(.*)\\]";           // 定义正则规则

        Pattern reg = Pattern.compile(patten);  // 编译成正则对象
        Matcher matcher = reg.matcher(text);    // 去text中进行匹配

        String jsonStr = "";
        // 如果text中的内容和正则规则匹配上就取出来赋值给jsonStr变量
        if (matcher.find()) {
            jsonStr = matcher.group(0);
            System.out.println("jsonStr = " + jsonStr);
        } else {
            System.out.println("no match");
        }

        // 4.对json数据进行更近一步的解析
        //  4.1.将第一层json(省份数据)解析为JavaBean
        List<CovidBean> pCovidBeans = JSON.parseArray(jsonStr, CovidBean.class);

        for (CovidBean pBean : pCovidBeans) {
            // System.out.println(pBean);
            // 先设置时间字段
            pBean.setDatetime(datetime);
            // 获取cities 和 dangerAreas
            String citiesStr = pBean.getCities();
            String dangerAreasStr = pBean.getDangerAreas();

            // 5.将第二层json(城市数据, 高危地区) 解析为JavaBean
            List<CovidBean> cityBeans = JSON.parseArray(citiesStr, CovidBean.class);
            for (CovidBean bean : cityBeans) {
                bean.setDatetime(datetime);
                bean.setPid(pBean.getLocationId()); // 把省份的id作为城市的pid
                bean.setProvinceShortName(pBean.getProvinceShortName());

                // System.out.println("bean = " + bean);
                // 后续需要将城市疫情数据发送给Kafka
                // 将JavaBean转为jsonStr再发送给Kafka
                String beanStr = JSON.toJSONString(bean);
                kafkaTemplate.send("covid19", bean.getPid(), beanStr);
            }

            List<DangerAreasBean> dangerAreasBeans = JSON.parseArray(dangerAreasStr, DangerAreasBean.class);
            // for (DangerAreasBean areasBean : dangerAreasBeans) {
            //     System.out.println(areasBean);
            // }

            // 6.获取第一层json(省份数据)中的每一天的统计数据
            String statisticsDataUrl = pBean.getStatisticsData();
            String statisticsDataStr = HttpUtils.getHtml(statisticsDataUrl);

            // 获取statisticsDataStr中的data字段对应的数据
            JSONObject jsonObject = JSON.parseObject(statisticsDataStr);
            String dataStr = jsonObject.getString("data");


            // 将爬取解析出来的每一天的数据设置回省份pBean中的statisticsData字段中(之前该字段只是一个URL)
            pBean.setStatisticsData(dataStr);
            pBean.setCities(null);
            pBean.setDangerAreas(null);

            // System.out.println("pBean = " + pBean);
            // 后续需要将省份疫情数据发送给Kafka
            String pBeanStr = JSON.toJSONString(pBean);

            /**
             * bug点
             * org.apache.kafka.common.errors.SerializationException: Can't convert key of class java.lang.String to class org.apache.kafka.common.serialization.IntegerSerializer specified in key.serializer
             * KafkaProducerConfig中设置了 key序列化类型为 IntegerSerializer !
             */
            kafkaTemplate.send("covid19", pBean.getLocationId(), pBeanStr);
        }
    }
}
